package com.zhd.listener;

import java.math.BigDecimal;
import java.util.List;
import java.util.Map;

import org.apache.commons.codec.binary.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.zhd.utils.FormulaCalculateUtil;

/**
 * 消息由canal发的消息
 * @author Administrator
 *
 */
@Component
@RocketMQMessageListener(topic = "iot-canal-gujun-data", consumerGroup = "iot-canal-gujun-data")
public class CanalListener implements RocketMQListener<String> {
	private Logger logger = LoggerFactory.getLogger(this.getClass());
	
	@Autowired
    private ObjectMapper objectMapper;
	@Autowired
    private RocketMQTemplate rocketMQTemplate;
	
	@SuppressWarnings("unchecked")
	@Override
	public void onMessage(String message) {
		try {
			logger.info("canal mq数据message={}", message);
			
			Map<String, Object> messageMap = objectMapper.readValue(message, Map.class);
			// 表名
			String table = messageMap.getOrDefault("table", "") + "";
			// sql类型，只处理type=INSERT的数据
			String type = messageMap.getOrDefault("type", "") + "";
			logger.info("解析message：table={}，type={}", table, type);
			if(!"INSERT".equalsIgnoreCase(type)) {
				logger.info("type={}，系统不处理该类型业务，只处理type=INSERT业务");
				return;
			}
			
			// 数据处理
			List<Map<String, Object>> canalDataList = (List<Map<String, Object>>) messageMap.get("data");
			if(canalDataList == null || canalDataList.size() == 0) {
				logger.info("data数据为空");
				return;
			}
			
			if("st_river_r".equalsIgnoreCase(table)) {// 水位、大坝安全
				logger.info("table={}", table);
				// 一条数据为一个采集器加上传感器的数据
				Map<String, Object> deviceMap = null;
				String key = null;
				Object val = null;
				List<Map<String, Object>> dataList = null;
				Map<String, Object> dataMap = null;
				for(Map<String, Object> map : canalDataList) {
					// 组装传感器数据
					dataList = Lists.newArrayList();
					
					// 动态解析所有字段
					for (Map.Entry<String, Object> entry : map.entrySet()) {
						key = entry.getKey().toLowerCase();
						val = entry.getValue();
						
						if(StringUtils.equals("sl01", key)) {// 渗流需要单独计算
							BigDecimal shenliu = FormulaCalculateUtil.shenliu(val, 0.24);
							val = shenliu;
						}
						
						dataMap = Maps.newHashMap();
						dataMap.put("id", key);
						dataMap.put("key", key);
						dataMap.put("val", val);
						dataList.add(dataMap);
					}
					
					/*dataMap.put("id", "z01");
					dataMap.put("key", "z01");
					dataMap.put("val", map.get("Z01"));
					data.add(dataMap);
					
					dataMap = Maps.newHashMap();
					dataMap.put("id", "z02");
					dataMap.put("key", "z02");
					dataMap.put("val", map.get("Z02"));
					data.add(dataMap);
					
					dataMap = Maps.newHashMap();
					dataMap.put("id", "z03");
					dataMap.put("key", "z03");
					dataMap.put("val", map.get("Z03"));
					data.add(dataMap);
					
					dataMap = Maps.newHashMap();
					dataMap.put("id", "z04");
					dataMap.put("key", "z04");
					dataMap.put("val", map.get("Z04"));
					data.add(dataMap);
					
					// 葛南渗流计，需公式计算
					BigDecimal shenliu = FormulaCalculateUtil.shenliu(map.get("Z05"), 0.24);
					dataMap = Maps.newHashMap();
					dataMap.put("id", "z05");
					dataMap.put("key", "z05");
					dataMap.put("val", shenliu);
					data.add(dataMap);
					
					dataMap = Maps.newHashMap();
					dataMap.put("id", "z06");
					dataMap.put("key", "z06");
					dataMap.put("val", map.get("Z06"));
					data.add(dataMap);
					
					dataMap = Maps.newHashMap();
					dataMap.put("id", "z07");
					dataMap.put("key", "z07");
					dataMap.put("val", map.get("Z07"));
					data.add(dataMap);
					
					dataMap = Maps.newHashMap();
					dataMap.put("id", "z08");
					dataMap.put("key", "z08");
					dataMap.put("val", map.get("Z08"));
					data.add(dataMap);
					
					dataMap = Maps.newHashMap();
					dataMap.put("id", "z09");
					dataMap.put("key", "z09");
					dataMap.put("val", map.get("Z09"));
					data.add(dataMap);
					
					dataMap = Maps.newHashMap();
					dataMap.put("id", "z10");
					dataMap.put("key", "z10");
					dataMap.put("val", map.get("Z10"));
					data.add(dataMap);*/
					
					// 组装采集器数据
					deviceMap = Maps.newHashMap();
					deviceMap.put("devId", map.get("STCD"));
					deviceMap.put("sn", map.get("STCD"));
					deviceMap.put("time", map.get("TM"));
					deviceMap.put("tenantId", null);
					deviceMap.put("tags", null);
					deviceMap.put("data", dataList);
					
					String deviceMessage = objectMapper.writeValueAsString(deviceMap);
					
					rocketMQTemplate.asyncSend("iot-gujun", MessageBuilder.withPayload(deviceMessage).setHeader("KEYS", map.get("STCD")).build(), new SendCallback() {
						@Override
					    public void onSuccess(SendResult result) {
					        if (result.getSendStatus() == SendStatus.SEND_OK) {
					        	logger.info("vc其他数据转发成功={}", deviceMessage);
					        }
					    }
					    @Override
					    public void onException(Throwable e) {
					    	logger.error("vc其他数据转发失败={}", e.getMessage());
					    }
					});
				}
			} else if("st_pptn_r".equalsIgnoreCase(table)) {// 雨量
				logger.info("table={}", table);
				// 一条数据为一个采集器加上传感器的数据
				Map<String, Object> deviceMap = null;
				for(Map<String, Object> map : canalDataList) {
					String intv = map.get("INTV") + "";
					BigDecimal bintv = new BigDecimal(intv);
					if(new BigDecimal(1).compareTo(bintv) == 0) {
						logger.info("时段INTV={}，不予处理", intv);
						continue;
					}
					
					// 组装传感器数据
					List<Map<String, Object>> dataList = Lists.newArrayList();
					Map<String, Object> dataMap = Maps.newHashMap();
					dataMap.put("id", "drp");
					dataMap.put("key", "drp");
					dataMap.put("val", map.get("DRP"));
					dataList.add(dataMap);
					
					// 组装采集器数据
					deviceMap = Maps.newHashMap();
					deviceMap.put("devId", map.get("STCD"));
					deviceMap.put("sn", map.get("STCD"));
					deviceMap.put("time", map.get("TM"));
					deviceMap.put("tenantId", null);
					deviceMap.put("tags", null);
					deviceMap.put("data", dataList);
					
					String deviceMessage = objectMapper.writeValueAsString(deviceMap);
					
					rocketMQTemplate.asyncSend("iot-gujun", MessageBuilder.withPayload(deviceMessage).setHeader("KEYS", map.get("STCD")).build(), new SendCallback() {
						@Override
		                public void onSuccess(SendResult result) {
		                    if (result.getSendStatus() == SendStatus.SEND_OK) {
		                        logger.info("vc雨量数据转发成功={}", deviceMessage);
		                    }
		                }
		                @Override
		                public void onException(Throwable e) {
		                    logger.error("vc雨量数据转发失败={}", e.getMessage());
		                }
		            });
				}
			} else {
				logger.info("未找到table={}对应的业务处理", table);
			}
		} catch (JsonMappingException e) {
			logger.error("JsonMappingException" , e);
		} catch (JsonProcessingException e) {
			logger.error("JsonProcessingException" , e);
		}
	}
}